package fun.tan90.easy.binlog.worker.handler;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import fun.tan90.easy.binlog.base.constants.EasyBinlogConstants;
import fun.tan90.easy.binlog.base.model.OpenBinlogClientReq;
import fun.tan90.easy.binlog.base.strategy.AbstractExecuteStrategy;
import fun.tan90.easy.binlog.base.threadpool.EasyBinlogThreadPool;
import fun.tan90.easy.binlog.worker.cache.BinaryLogClientKey;
import fun.tan90.easy.binlog.worker.cache.DataSourceKey;
import fun.tan90.easy.binlog.worker.handler.listener.BinlogEventListener;
import fun.tan90.easy.binlog.worker.handler.listener.BinlogLifecycleListener;
import fun.tan90.easy.binlog.worker.handler.processor.BinlogEventProcessor;
import lombok.extern.slf4j.Slf4j;
import org.noear.socketd.transport.core.Entity;
import org.noear.socketd.transport.core.Message;
import org.noear.socketd.transport.core.entity.StringEntity;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.CompletableFuture;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/12/15 19:42
 */
@Slf4j
@Component
public class TurnOnBinlogHandler implements AbstractExecuteStrategy<Message, Entity> {

    @Resource
    BinlogEventProcessor binlogEventProcessor;

    @Override
    public String mark() {
        return EasyBinlogConstants.TURN_ON_BINLOG;
    }

    @Override
    public Entity executeResp(Message message) {
        OpenBinlogClientReq openBinlogClientReq = JSONUtil.toBean(message.dataAsString(), OpenBinlogClientReq.class);
        String masterCode = openBinlogClientReq.getMasterCode();
        String username = openBinlogClientReq.getUsername();
        String password = openBinlogClientReq.getPassword();
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (!DataSourceKey.hasDataSource(masterCode)) {
                HikariConfig config = new HikariConfig();
                config.setJdbcUrl(StrUtil.format(StrUtil.format("jdbc:mysql://{}:{}", openBinlogClientReq.getHostname(), openBinlogClientReq.getPort())));
                config.setUsername(username);
                config.setPassword(password);
                config.setMinimumIdle(4);
                config.setMaximumPoolSize(20);
                config.setMaxLifetime(1800000L);
                HikariDataSource hikariDataSource = new HikariDataSource(config);
                try (Connection connection = hikariDataSource.getConnection()) {
                    try (PreparedStatement statement = connection.prepareStatement("select version() as version")) {
                        try (ResultSet resultSet = statement.executeQuery()) {
                            if (resultSet.next()) {
                                String version = resultSet.getString(1);
                                log.info("masterCode {}, version {}", masterCode, version);
                                DataSourceKey.addDataSource(masterCode, hikariDataSource);
                            }
                        }
                    }
                } catch (Exception e) {
                    log.error(e.getMessage());
                    return "0";
                }
            }
            return "1";
        }, EasyBinlogThreadPool.EB_FIXED_EXECUTOR).thenApply(res -> {
            if ("0".equals(res)) {
                return "00";
            }
            if (!BinaryLogClientKey.hasBinaryLogClient(masterCode)) {
                BinaryLogClient client = new BinaryLogClient(openBinlogClientReq.getHostname(), openBinlogClientReq.getPort(), username, password);
                client.setServerId(openBinlogClientReq.getServerId());
                client.setKeepAlive("1".equals(openBinlogClientReq.getKeepAlive()));
                client.setKeepAliveInterval(openBinlogClientReq.getKeepAliveInterval());
                client.setHeartbeatInterval(openBinlogClientReq.getHeartbeatInterval());
                client.setConnectTimeout(openBinlogClientReq.getConnectTimeout());

                String filename = openBinlogClientReq.getFilename();
                Long position = openBinlogClientReq.getPosition();

                if (StringUtils.hasLength(filename) && !ObjectUtils.isEmpty(position)) {
                    client.setBinlogFilename(filename);
                    client.setBinlogPosition(position);
                }
                client.registerEventListener(new BinlogEventListener(masterCode, openBinlogClientReq.getOffsetMilliseconds(), binlogEventProcessor));
                client.registerLifecycleListener(new BinlogLifecycleListener(masterCode, binlogEventProcessor));
                try {
                    client.connect(6000);
                    BinaryLogClientKey.addBinaryLogClient(masterCode, client);
                } catch (Exception e) {
                    log.error(e.getMessage());
                    return res + "0";
                }
            }
            return res + "1";
        });
        return new StringEntity(future.join());
    }
}
